#include <string>

#include <Core/MySQL/MySQLClient.h>
#include <Core/MySQL/Authentication.h>
#include <Core/MySQL/PacketsGeneric.h>
#include <Core/MySQL/PacketsConnection.h>
#include <Core/MySQL/PacketsProtocolText.h>
#include <Core/MySQL/PacketsReplication.h>
#include <IO/ReadBufferFromString.h>
#include <IO/WriteBufferFromString.h>

#include <boost/program_options.hpp>

int main(int argc, char ** argv)
{
    using namespace DB;
    using namespace MySQLProtocol;
    using namespace MySQLProtocol::Generic;
    using namespace MySQLProtocol::Authentication;
    using namespace MySQLProtocol::ConnectionPhase;
    using namespace MySQLProtocol::ProtocolText;


    uint8_t sequence_id = 1;
    String user = "default";
    String password = "123";
    String database;

    UInt8 charset_utf8 = 33;
    UInt32 max_packet_size = MAX_PACKET_LENGTH;
    String mysql_native_password = "mysql_native_password";

    UInt32 server_capability_flags = CLIENT_PROTOCOL_41 | CLIENT_SECURE_CONNECTION | CLIENT_PLUGIN_AUTH
        | CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA | CLIENT_CONNECT_WITH_DB | CLIENT_DEPRECATE_EOF;

    UInt32 client_capability_flags = CLIENT_PROTOCOL_41 | CLIENT_PLUGIN_AUTH | CLIENT_SECURE_CONNECTION;

    /// Handshake packet
    {
        /// 1. Greeting:
        /// 1.1 Server writes greeting to client
        std::string s0;
        WriteBufferFromString out0(s0);

        Handshake server_handshake(server_capability_flags, -1, "ClickHouse", "mysql_native_password", "aaaaaaaaaaaaaaaaaaaaa", CharacterSet::utf8_general_ci);
        server_handshake.writePayload(out0, sequence_id);

        /// 1.2 Client reads the greeting
        ReadBufferFromString in0(s0);
        Handshake client_handshake;
        client_handshake.readPayload(in0, sequence_id);

        /// Check packet
        ASSERT(server_handshake.capability_flags == client_handshake.capability_flags)
        ASSERT(server_handshake.status_flags == client_handshake.status_flags)
        ASSERT(server_handshake.server_version == client_handshake.server_version)
        ASSERT(server_handshake.protocol_version == client_handshake.protocol_version)
        ASSERT(server_handshake.auth_plugin_data.substr(0, 20) == client_handshake.auth_plugin_data)
        ASSERT(server_handshake.auth_plugin_name == client_handshake.auth_plugin_name)

        /// 2. Greeting Response:
        std::string s1;
        WriteBufferFromString out1(s1);

        /// 2.1 Client writes to server
        Native41 native41(password, client_handshake.auth_plugin_data);
        String auth_plugin_data = native41.getAuthPluginData();
        HandshakeResponse client_handshake_response(
            client_capability_flags, max_packet_size, charset_utf8, user, database, auth_plugin_data, mysql_native_password);
        client_handshake_response.writePayload(out1, sequence_id);

        /// 2.2 Server reads the response
        ReadBufferFromString in1(s1);
        HandshakeResponse server_handshake_response;
        server_handshake_response.readPayload(in1, sequence_id);

        /// Check
        ASSERT(server_handshake_response.capability_flags == client_handshake_response.capability_flags)
        ASSERT(server_handshake_response.character_set == client_handshake_response.character_set)
        ASSERT(server_handshake_response.username == client_handshake_response.username)
        ASSERT(server_handshake_response.database == client_handshake_response.database)
        ASSERT(server_handshake_response.auth_response == client_handshake_response.auth_response)
        ASSERT(server_handshake_response.auth_plugin_name == client_handshake_response.auth_plugin_name)
    }

    /// OK Packet
    {
        // 1. Server writes packet
        std::string s0;
        WriteBufferFromString out0(s0);
        OKPacket server(0x00, server_capability_flags, 0, 0, 0, "", "");
        server.writePayload(out0, sequence_id);

        // 2. Client reads packet
        ReadBufferFromString in0(s0);
        ResponsePacket client(server_capability_flags);
        client.readPayload(in0, sequence_id);

        // Check
        ASSERT(client.getType() == PACKET_OK)
        ASSERT(client.ok.header == server.header)
        ASSERT(client.ok.status_flags == server.status_flags)
        ASSERT(client.ok.capabilities == server.capabilities)
    }

    /// ERR Packet
    {
        // 1. Server writes packet
        std::string s0;
        WriteBufferFromString out0(s0);
        ERRPacket server(123, "12345", "This is the error message");
        server.writePayload(out0, sequence_id);

        // 2. Client reads packet
        ReadBufferFromString in0(s0);
        ResponsePacket client(server_capability_flags);
        client.readPayload(in0, sequence_id);

        // Check
        ASSERT(client.getType() == PACKET_ERR)
        ASSERT(client.err.header == server.header)
        ASSERT(client.err.error_code == server.error_code)
        ASSERT(client.err.sql_state == server.sql_state)
        ASSERT(client.err.error_message == server.error_message)
    }

    /// EOF Packet
    {
        // 1. Server writes packet
        std::string s0;
        WriteBufferFromString out0(s0);
        EOFPacket server(1, 1);
        server.writePayload(out0, sequence_id);

        // 2. Client reads packet
        ReadBufferFromString in0(s0);
        ResponsePacket client(server_capability_flags);
        client.readPayload(in0, sequence_id);

        // Check
        ASSERT(client.getType() == PACKET_EOF)
        ASSERT(client.eof.header == server.header)
        ASSERT(client.eof.warnings == server.warnings)
        ASSERT(client.eof.status_flags == server.status_flags)
    }

    /// ColumnDefinition Packet
    {
        // 1. Server writes packet
        std::string s0;
        WriteBufferFromString out0(s0);
        ColumnDefinition server("schema", "tbl", "org_tbl", "name", "org_name", 33, 0x00, MYSQL_TYPE_STRING, 0x00, 0x00);
        server.writePayload(out0, sequence_id);

        // 2. Client reads packet
        ReadBufferFromString in0(s0);
        ColumnDefinition client;
        client.readPayload(in0, sequence_id);

        // Check
        ASSERT(client.column_type == server.column_type)
        ASSERT(client.column_length == server.column_length)
        ASSERT(client.next_length == server.next_length)
        ASSERT(client.character_set == server.character_set)
        ASSERT(client.decimals == server.decimals)
        ASSERT(client.name == server.name)
        ASSERT(client.org_name == server.org_name)
        ASSERT(client.table == server.table)
        ASSERT(client.org_table == server.org_table)
        ASSERT(client.schema == server.schema)
    }

    {
        try
        {
            boost::program_options::options_description desc("Allowed options");
            desc.add_options()("host", boost::program_options::value<std::string>()->required(), "master host")(
                "port", boost::program_options::value<std::int32_t>()->required(), "master port")(
                "user", boost::program_options::value<std::string>()->required(), "master user")(
                "password", boost::program_options::value<std::string>()->required(), "master password")(
                "db", boost::program_options::value<std::string>()->required(), "replicate do db");

            boost::program_options::variables_map options;
            boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), options);
            if (argc == 0)
            {
                return 1;
            }

            auto host = options.at("host").as<DB::String>();
            auto port = options.at("port").as<DB::Int32>();
            auto master_user = options.at("user").as<DB::String>();
            auto master_password = options.at("password").as<DB::String>();
            auto replicate_db = options.at("db").as<DB::String>();

            std::cerr << "Master Host: " << host << ", Port: " << port << ", User: " << master_user << ", Password: " << master_password
                      << ", Replicate DB: " << replicate_db << std::endl;

            UInt32 slave_id = 9004;
            MySQLClient slave(host, port, master_user, master_password);

            /// Connect to the master.
            slave.connect();

            ///  start to dump binlog.
            slave.startBinlogDump(slave_id, replicate_db, "", 4);

            /// Read one binlog event on by one.
            while (true)
            {
                auto event = slave.readOneBinlogEvent();
                switch (event->type())
                {
                    case MYSQL_QUERY_EVENT: {
                        auto binlog_event = std::static_pointer_cast<QueryEvent>(event);
                        binlog_event->dump(std::cerr);

                        Position pos = slave.getPosition();
                        std::cerr << "Binlog Name: " << pos.binlog_name << ", Pos: " << pos.binlog_pos << std::endl;
                        break;
                    }
                    case MYSQL_WRITE_ROWS_EVENT: {
                        auto binlog_event = std::static_pointer_cast<WriteRowsEvent>(event);
                        binlog_event->dump(std::cerr);

                        Position pos = slave.getPosition();
                        std::cerr << "Binlog Name: " << pos.binlog_name << ", Pos: " << pos.binlog_pos << std::endl;
                        break;
                    }
                    case MYSQL_UPDATE_ROWS_EVENT: {
                        auto binlog_event = std::static_pointer_cast<UpdateRowsEvent>(event);
                        binlog_event->dump(std::cerr);

                        Position pos = slave.getPosition();
                        std::cerr << "Binlog Name: " << pos.binlog_name << ", Pos: " << pos.binlog_pos << std::endl;
                        break;
                    }
                    case MYSQL_DELETE_ROWS_EVENT: {
                        auto binlog_event = std::static_pointer_cast<DeleteRowsEvent>(event);
                        binlog_event->dump(std::cerr);

                        Position pos = slave.getPosition();
                        std::cerr << "Binlog Name: " << pos.binlog_name << ", Pos: " << pos.binlog_pos << std::endl;
                        break;
                    }
                    default:
                        break;
                }
            }
        }
        catch (const Exception & ex)
        {
            std::cerr << "Error: " << ex.message() << std::endl;
            return 1;
        }
    }
}
